package com.cx.mall.demo.controller;

import com.cx.mall.common.util.ResponseVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;

@Slf4j
@RestController
@RequestMapping("/mq")
public class MQController {
    @Autowired
    StreamBridge streamBridge;  //消息发送器
    @GetMapping("/send")
    public ResponseVo send(@RequestParam String msg){
        //发送消息 到testOut-out-0 通道
        streamBridge.send("testOut-out-0",msg);
        return ResponseVo.success("发送成功");
    }

    @GetMapping("/payload")
    public ResponseVo payLoad(@RequestParam String msg){
        Message<?> message = MessageBuilder.withPayload(msg)
                .setHeader("TAG", "success") // 定义消息头传递 Tag
                .build();
        streamBridge.send("testOutput-out-0", message);
        return ResponseVo.success("发送成功");
    }

    @GetMapping("/delay")
    public ResponseVo delay(@RequestParam String msg){
        Message<?> message = MessageBuilder.withPayload(msg)
                .setHeader("DELAY", 5)
                .setHeader("TAG", "delay")
                .build();
        streamBridge.send("testOutput-out-0", message);
        return ResponseVo.success("发送成功");
    }
    // 顺序消息
    // 同一个订单号数据进行同一个队列
    @GetMapping("/order")
    public ResponseVo order(String msg,String orderId){
        Message<?> message = MessageBuilder.withPayload(msg)
                // 同一个订单号数据进行同一个队列
                .setHeader("ORDERLY_KEY", orderId)
                .build();
        streamBridge.send("testOutput-out-0", message);
        return ResponseVo.success("发送成功");
    }

    @Autowired // 注入RedisTemplate接口
    private RedisTemplate redisTemplate;

    @GetMapping("/unique")
    public ResponseVo unique(String msg,String key) {
        Message<?> message = MessageBuilder.withPayload(msg).build();

        // 必须设置合理过期时间,未设置过期时间导致锁无法自动释放(如系统崩溃)
        if (redisTemplate.opsForValue().setIfAbsent("REQ_" + key, "1", 10, TimeUnit.SECONDS)) {
           log.info("发送唯一消息" + message);
            streamBridge.send("test3Output-out-0", message);
        }
        return ResponseVo.success("唯一消息发送成功" );
    }
}